package com.wis.bolt.risk;

import com.wis.TopologyDrive;
import org.apache.commons.lang.StringUtils;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * @Description: traffic_wis
 * @author: fan
 * @Date: Created in 2018/12/19 11:14
 * @Modified By:
 */
public class ZongHeZhuanBolt extends BaseRichBolt {
    private OutputCollector collector;
    private static String typeProduce;
    private static Queue tupleQueue = new ConcurrentLinkedQueue();
    private static Queue tupleIdQueue = new ConcurrentLinkedQueue();

    @Override
    public void prepare(Map map, TopologyContext context, OutputCollector collector) {
        this.collector = collector;
        typeProduce = map.get(TopologyDrive.TYPE_PRODUCE).toString();
    }

    @Override
    public void execute(Tuple tuple) {
        String key = tuple.getStringByField("key");
        String value = tuple.getStringByField("value");

        try {
            if (value.equals("over")) {
                if (typeProduce.equals("1")) {
                    this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", "00") + "00", "over", "over"));
                } else if (typeProduce.equals("2")) {
                    this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", "40") + "00", "over", "over"));
                } else if (typeProduce.equals("3")) {
                    this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", "80") + "00", "over", "over"));
                }
            } else if (StringUtils.isNotBlank(value)/* && value.length() > 4*/) {
                String[] maxVlues = value.split(",", -1);
                String[] ids = tuple.getStringByField("id").split(",", -1);
                for (int i = 0; i < maxVlues[0].split("_", -1).length - 1; i++) {// 控制有多少个时间点
                    for (int j = 0; j < maxVlues.length; j++) {// 控制有多少行
                        // if (ids[j].contains("G1212_324")) {
                        //     System.out.println("-----");
                        // }
                        String tmp = maxVlues[j].split("_", -1)[i];
                        // 如果最大值的参数大于0，则添加
                        if (Integer.parseInt(tmp) > 0) {
                            tupleQueue.add(tmp);
                            tupleIdQueue.add(ids[j]);
                        }
                    }
                    // 每个list值都是每个时间最大值非0 拼成的
                    if (tupleQueue.size() > 0)
                        if (typeProduce.equals("1")) {
                            this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", "00") + "00", StringUtils.join(tupleIdQueue, ","), StringUtils.join(tupleQueue, ",")));
                        } else if (typeProduce.equals("2")) {
                            this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", (i + 1) + "0") + "00", StringUtils.join(tupleIdQueue, ","), StringUtils.join(tupleQueue, ",")));
                        } else if (typeProduce.equals("3")) {
                            if (i < 9) {
                                this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", "0" + (i + 1)) + "00", StringUtils.join(tupleIdQueue, ","), StringUtils.join(tupleQueue, ",")));
                            } else {
                                this.collector.emit(new Values("1" + typeProduce + "0" + key.replace("sc", (i + 1) + "") + "00", StringUtils.join(tupleIdQueue, ","), StringUtils.join(tupleQueue, ",")));
                            }
                        }
                    tupleQueue.clear();
                    tupleIdQueue.clear();
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            this.collector.ack(tuple);
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("key", "id", "value"));
    }
}
